package com.asiainfo.dacp.datastash.broker.facade.service;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitStepShardMeta;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnitStep;
import com.asiainfo.dacp.metamodel.domain.ds.model.MetaDataSource;
import com.asiainfo.dacp.util.StringUtils;
import com.asiainfo.dacp.util.vfs.VfsUtils;
import com.google.common.base.Strings;

import org.apache.commons.vfs2.FileObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

/**
 * 分片节点的服务
 *
 * @author MeiKefu
 * @date 2017-12-14
 */
@Service
public class UnitShardingService {

    @Autowired
    private StashTraceLogger tracelog;
    private String clzName = this.getClass().getSimpleName();

    public List<List<Map<String,String>>> sharding(DataStashUnitStep stepDomain, StashUnit stashUnit){

        int nShardCnt = (Integer)stepDomain.getStepCfgValMap().get("shardCnt");

        List<Map<String,Object>> contents = (List<Map<String,Object>>)stepDomain.getStepCfgValMap().get("contents");

        String strg = (String)stepDomain.getStepCfgValMap().get("strg");

        List<List<Map<String,String>>> shardContents = new ArrayList<>(nShardCnt);

        if(contents==null || contents.size()==0){
            tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit,clzName,"sharding","分片组件配置错误"));
            return shardContents;
        }

        List<StashUnitStepShardMeta> fileContents = new ArrayList<>();

        for(Map<String,Object> cont : contents){
            String dsName = (String)cont.get("dsName");
            String contentStr = (String)cont.get("content");
            StashUnitStepShardMeta shardMeta = new StashUnitStepShardMeta(dsName,contentStr,strg);
            String filter="";;
            if(cont.get("filter")!=null)
            {
                filter=StringUtils.variableSubstitution((String)cont.get("filter"),stashUnit);
            }
            shardMeta.setFilter(filter);
            fileContents.add(shardMeta);
        }

        List<Map<String,String>> elements = discovery(fileContents,stashUnit);

        int eleCnt = elements.size()/nShardCnt+ (elements.size()%nShardCnt==0?0:1);

        List<Map<String,String>> pieceMap = null;
        for(int i=0; i < elements.size(); i++){
            if(i%eleCnt==0){
                pieceMap = new ArrayList<>();
                shardContents.add(pieceMap);
            }
            pieceMap.add(elements.get(i));
        }

        return shardContents;
    }

    private List<Map<String,String>> discovery(List<StashUnitStepShardMeta> fileContents,StashUnit stashUnit) {

        tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit,clzName,"discoveryFile","发现数据内容"));
        List<Map<String,String>> result = new ArrayList<>();
        //TODO 需要实现多种数据文件的识别

        //1.目录下面，或目录的的正则
        //2.打开状态文件，按照分割
        for (StashUnitStepShardMeta shardMeta:fileContents){

            if("file".equals(shardMeta.getStrg()) || "hdfs".equals(shardMeta.getStrg())){
                FileObject stateFilePath =  null;
                try { 
                    MetaDataSource dataSource = BrokerContext.getInstance().getDsDefs().get(shardMeta.getDsName());

                    if( "hdfs".equals(shardMeta.getStrg()) && !Strings.isNullOrEmpty(dataSource.getDsAcct())){
                		System.setProperty("HADOOP_USER_NAME", dataSource.getDsAcct());
                	}
                    
                    String vfsPath= dataSource.getVfsHeader()+ StringUtils.variableSubstitution(shardMeta.getContent(), stashUnit);
                    stateFilePath = VfsUtils.getInstance().resolveFile(vfsPath);

                    if(stateFilePath.isFile()){
                        result.add(getContent(shardMeta.getDsName(),stateFilePath.getName().getPath()));
                    }else if(stateFilePath.isFolder()){

                        FileObject[] fileObjects = stateFilePath.getChildren();

                        if(fileObjects!=null && fileObjects.length>0){
                            Pattern pattern = null;

                            //TODO 需要实现目录下面的正则匹配
                            if(org.apache.commons.lang3.StringUtils.isNotBlank(shardMeta.getFilter())){
                                pattern = Pattern.compile(shardMeta.getFilter());
                            }

                            for (FileObject file : fileObjects) {
                                String pathName = file.getName().getPath();
                                String fileName = file.getName().getBaseName(); 
                                if(pattern == null || pattern.matcher(fileName).matches()){
                                    result.add(getContent(shardMeta.getDsName(),pathName));
                                }
                            }
                        }else{
                            tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"scheduleData.vfsPath()",vfsPath+"未目录下面没有文件"));
                        }
                    }else{
                        tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"scheduleData.vfsPath()",vfsPath+"目录不存在"));
                    }
                } catch (Exception e) {
                    tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"discoveryFile:error","查找数据内容失败"+e.getMessage()));
                }finally {
                    VfsUtils.getInstance().closeFileObject(stateFilePath);
                }

            }else if("sql".equals(shardMeta.getStrg())){
                //TODO SQL分片还没有完成
                result.add(getSql(shardMeta.getDsName(),shardMeta.getContent()));
            }
        }
        tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"discoveryFile:complete","成功数据内容发现，分片数据"+result.size()));

        return result;
    }

    private Map<String,String> getContent(String dsName,String contentStr){
        Map<String,String> content = new HashMap<>();
        content.put("dsName",dsName);
        content.put("content",contentStr);
        return content;
    }

    private Map<String,String> getSql(String dsName,String sql){
        Map<String,String> content = new HashMap<>();
        content.put("dsName",dsName);
        content.put("sql",sql);
        return content;
    }

}
